import com.nekonomics.lakehouse.common.FlinkRunner._
import com.nekonomics.lakehouse.common.util.SqlUtil._
import com.nekonomics.lakehouse.common.adt.PaymentStatus._

/**
 *
 * 业务过程为订单，
 * sku粒度，
 * 度量值为已支付订单的销售金额和销售数量，
 * 统计周期为天
 *
 */
object OrderSkuSaleDailyDws {
  def main(args: Array[String]): Unit = {

    implicit val config: SqlRunnerConfig = SqlRunnerConfig(groupId = "order_sku_sale_daily_dws", port = 8085, ckp = true)
    runSql { (_, tableEnv) =>
      tableEnv.executeSql(
        sql"""
             |CREATE TABLE IF NOT EXISTS dws.order_sku_sale_daily_dws (
             |  sku_id BIGINT,
             |  spu_id BIGINT,
             |  merchant_id BIGINT,
             |  date_key DATE,
             |  product_name STRING,
             |  total_sales DECIMAL(10, 2),
             |  total_quantity INT,
             |  PRIMARY KEY (sku_id, date_key) NOT ENFORCED
             |)
             |""".stripMargin
      )

      tableEnv.executeSql(
        sql"""
             |INSERT INTO dws.order_sku_sale_daily_dws
             |SELECT
             |  oi.sku_id,
             |  MAX(oi.spu_id),
             |  MAX(oi.merchant_id),
             |  CAST(DATE_FORMAT(oi.order_created_at, 'yyyy-MM-dd') AS DATE),
             |  MAX(p.product_name),
             |  SUM(oi.total_amount),
             |  SUM(oi.quantity)
             |FROM dwd.order_item_detail_dwd oi
             |JOIN dim.product_sku_dim p ON oi.sku_id = p.sku_id
             |WHERE oi.payment_status = '$Success'
             |GROUP BY
             |  oi.sku_id,
             |  DATE_FORMAT(oi.order_created_at, 'yyyy-MM-dd');
             |""".stripMargin
      )

    }
  }





}
